package com.xxx.config;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.xxx.handler.HelloEventHandler;
import com.xxx.pojo.HelloEventFactory;
import com.xxx.pojo.MessageModel;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Disruptor消息队列配置管理器
 */
@Configuration
public class MQManager {

    @Bean("messageModel")
    public RingBuffer<MessageModel> messageModelRingBuffer() {
        // 定义处理事件的线程池， Disruptor通过java.util.concurrent.ExecutorService提供的线程来触发consumer的事件处理
        ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

        // 指定事件工厂
        HelloEventFactory factory = new HelloEventFactory();

        // 指定ring buffer字节大小，必须为2的N次方（能将求模运算转为位运算提高效率），否则将影响效率
        int bufferSize = 1024 * 256;

        // 单线程模式，获取额外的性能
        Disruptor<MessageModel> disruptor = new Disruptor<>(
                factory,                        // 事件工厂
                bufferSize,                     // ring buffer size
                executor,                       // 处理事件的线程池
                ProducerType.SINGLE,            // 单线程模式
                new BlockingWaitStrategy());    // 阻塞等待策略

        // 设置事件业务处理器---消费者
        disruptor.handleEventsWith(new HelloEventHandler());

        // 启动disruptor线程  
        disruptor.start();

        // 获取ring buffer环，用于接取生产者生产的事件
        RingBuffer<MessageModel> ringBuffer = disruptor.getRingBuffer();

        return ringBuffer;
    }
}